package com.bw.gmall.realtime.dwd.db.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeOrderPaySucDetail extends BaseSqlApp {
    public static void main(String[] args) throws Exception {
        new DwdTradeOrderPaySucDetail().start(10016,
                4,
                Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);

    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {


        //1 读取Topic_db 数据
        readOdsDb(tableEnv,Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);
        // 2.过滤出支付成功数据 update 并且 status =1602
        Table paymentInfo = tableEnv.sqlQuery("select \n" +
                "    `data`['id'] id,\n" +
                "    `data`['order_id'] order_id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['payment_type'] payment_type,\n" +
                "    `data`['callback_time'] callback_time,\n" +
                "     row_time,\n" +
                "     proc_time,\n" +
                "     ts \n" +
                "from topic_db\n" +
                "where `database`='gmall'\n" +
                "and `table` = 'payment_info'\n" +
                "and `type` = 'update'\n" +
                "and `data`['payment_status'] = '1602'");
        tableEnv.createTemporaryView("payment_info",paymentInfo);
        // 3.读取Kafka的dwd订单数据
        tableEnv.executeSql("create table order_detail(\n" +
                "    id String,\n" +
                "    order_id  String,\n" +
                "    sku_id  String,\n" +
                "    user_id String,\n" +
                "    province_id String,\n" +
                "    activity_id String,\n" +
                "    activity_rule_id  String,\n" +
                "    coupon_id String, \n" +
                "    sku_name  String, \n" +
                "    order_price String, \n" +
                "    sku_num String, \n" +
                "    create_time String, \n" +
                "    split_total_amount  String, \n" +
                "    split_activity_amount String, \n" +
                "    split_coupon_amount String,\n" +
                "    ts  Bigint ,\n" +
                "    row_time as TO_TIMESTAMP_LTZ(ts * 1000,3),\n" +
                "    WATERMARK FOR row_time AS row_time - INTERVAL '5' SECOND \n" +
                ")"+ SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        // 4. 做Join INTERVAL Join,必须要用事件时间
        Table pay = tableEnv.sqlQuery("select\n" +
                "    p.id,\n" +
                "    o.order_id,\n" +
                "    p.user_id,\n" +
                "    payment_type ,\n" +
                "    callback_time payment_time,\n" +
                "    sku_id ,\n" +
                "    province_id,\n" +
                "    activity_id,\n" +
                "    activity_rule_id ,\n" +
                "    coupon_id, \n" +
                "    sku_name , \n" +
                "    order_price, \n" +
                "    sku_num, \n" +
                "    create_time, \n" +
                "    split_total_amount , \n" +
                "    split_activity_amount, \n" +
                "    split_coupon_amount,\n" +
                "    proc_time,\n" +
                "    p.ts \n" +
                "from payment_info p,order_detail o \n" +
                "where p.order_id = o.order_id \n" +
                "and p.row_time BETWEEN o.row_time - INTERVAL '15' MINUTE AND p.row_time + INTERVAL '5' SECOND");
        tableEnv.createTemporaryView("pay",pay);
        // 5、 跟字典表关联 Lookup Join
        createBaseDic(tableEnv);

        Table table = tableEnv.sqlQuery("select \n" +
                "    id,\n" +
                "    order_id,\n" +
                "    user_id,\n" +
                "    payment_type payment_type_code,\n" +
                "    dic_name payment_type_name,\n" +
                "    payment_time,\n" +
                "    sku_id ,\n" +
                "    province_id,\n" +
                "    activity_id,\n" +
                "    activity_rule_id ,\n" +
                "    coupon_id, \n" +
                "    sku_name , \n" +
                "    order_price, \n" +
                "    sku_num, \n" +
                "    create_time, \n" +
                "    split_total_amount , \n" +
                "    split_activity_amount, \n" +
                "    split_coupon_amount ,\n" +
                "    ts \n" +
                "from pay as p\n" +
                "join base_dic FOR SYSTEM_TIME AS OF p.proc_time as b\n" +
                "on p.payment_type = b.rowkey");

        // 6、 写入Kafka
        tableEnv.executeSql("create table "+Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS+"(\n" +
                "  id STRING,\n" +
                "  order_id STRING,\n" +
                "  user_id STRING,\n" +
                "  payment_type_cod STRING,  \n" +
                "  payment_type_name STRING, \n" +
                "  payment_time STRING,\n" +
                "  sku_id STRING, \n" +
                "  province_id STRING,\n" +
                "  activity_id STRING,\n" +
                "  activity_rule_id STRING, \n" +
                "  coupon_id STRING, \n" +
                "  sku_name STRING,  \n" +
                "  order_price STRING, \n" +
                "  sku_num STRING, \n" +
                "  create_time STRING, \n" +
                "  split_total_amount STRING,  \n" +
                "  split_activity_amount STRING, \n" +
                "  split_coupon_amount STRING,\n" +
                "  ts BIGINT\n" +
                ")"+SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS,Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));

        table.insertInto(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS).execute();
    }
}
